package com.nami.logic.rocketmq.consumer;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONUtil;
import com.nami.basic.bo.servicetrans.up.UpMsgWrap;
import com.nami.basic.constants.UpTopicConst;
import com.nami.basic.enums.DownMsgErrorCode;
import com.nami.basic.packet.Packet;
import com.nami.basic.packet.message.Message;
import com.nami.basic.packet.message.MsgUser;
import com.nami.logic.cache.conver.ConverCache;
import com.nami.logic.forward.TransDownMsgHandler;
import com.nami.logic.property.DownModeProperty;
import com.nami.logic.qos.QoSReceiveDaemonC2S;
import com.nami.logic.serializer.Serializer;
import com.nami.logic.service.ConverSyncHandler;
import com.nami.logic.service.DownMsgSender;
import com.nami.logic.service.PacketFactory;
import com.nami.logic.service.UserService;
import com.nami.logic.utils.ConverUtil;
import com.nami.store.api.bo.UserWrap;
import com.nami.store.api.request.conversation.ConverItem;
import com.nami.store.api.request.message.PrivateMsgReq;
import com.nami.store.api.service.RemoteConverService;
import com.nami.store.api.service.RemoteFriendService;
import com.nami.store.api.service.RemoteL2CacheService;
import com.nami.store.api.service.RemoteMessageService;
import com.nami.store.api.response.conversation.ConversationDTO;
import com.nami.store.api.response.user.UserDTO;
import com.nami.store.api.response.friend.FriendDTO;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.dromara.common.core.enums.conver.ConverTypeEnum;
import org.dromara.common.core.enums.user.UserStatusEnum;
import org.dromara.common.core.utils.ConverUtils;
import org.dromara.common.core.utils.IdentityUtil;
import org.dromara.common.json.utils.JsonUtils;
import org.jetbrains.annotations.NotNull;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;
import java.util.Map;

import static com.nami.basic.constants.CommonConst.COLON;
import static com.nami.logic.thread.ThreadPoolManager.PDISPATCHERS;


@Slf4j
@Component
@AllArgsConstructor
@RocketMQMessageListener(topic = UpTopicConst.C2C_TOPIC,
    consumerGroup = UpTopicConst.C2C_CONSUMER_GROUP,
    consumeMode = ConsumeMode.CONCURRENTLY)
public class PrivateMsgConsumer extends AbstractMessage implements RocketMQListener<String> {

    private final PacketFactory packetFactory;

    private final DownModeProperty downModeProperty;

    private final UserService userService;

    private final DownMsgSender sender;

    private final TransDownMsgHandler transDownMsgHandler;

    private final RemoteL2CacheService cacheService;

    private final ConverSyncHandler syncHandler;

    @DubboReference
    RemoteFriendService friendService;

    @DubboReference
    RemoteMessageService messageService;

    @DubboReference
    RemoteConverService converService;

    @Override
    public void onMessage(String msg) {
        try {
            TimeInterval timer = DateUtil.timer();
//            log.info("[c2c] begin. message:{}", msg);

            final UpMsgWrap upMsgWrap = JSONUtil.toBean(msg, UpMsgWrap.class);
            final Packet packet = upMsgWrap.getPacket();
            Message message = Serializer.prototype(packet.getSerial()).deserialize(packet.getBody(), Message.class);

            addSystemInfoToThisMsg(message);
            log.info("[c2c] 收到客户端【{}】发送的单聊消息, 消息体：{}", upMsgWrap.getUserIp(), JsonUtils.toJsonString(message));

            final String identityNo = IdentityUtil.buildIdentityNo(message.getSenderId(), upMsgWrap.getDeviceType());
            if (checkNullParams(upMsgWrap, message)) {
                sendErrorResp(sender, message, identityNo, upMsgWrap, DownMsgErrorCode.REQUEST_PARAMS_ERROR, packetFactory);
                log.warn("[c2c] 校验消息失败, 丢失核心参数：deviceType:{}, msg:{}", upMsgWrap.getDeviceType(), msg);
                return;
            } // end judge if

            Map<String, UserWrap> userMap = userService.getUserByIds(upMsgWrap.getBindingAppKey(), genSearchUserDTO(message));
            final UserWrap senderUser = userMap.get(message.getSenderId());
            final UserWrap recvUser = userMap.get(message.getReceiverId());

            if (null == senderUser || null == recvUser) {
                sendErrorResp(sender, message, identityNo, upMsgWrap, DownMsgErrorCode.USER_NOT_EXIST, packetFactory);
                return;
            } // end null if

            if (checkUserWasBan(upMsgWrap, senderUser, identityNo)) {
                sendErrorResp(sender, message, identityNo, upMsgWrap, DownMsgErrorCode.FROM_USER_BAN, packetFactory);
                return;
            } // end ban if

            if (checkToUserPermission(upMsgWrap, recvUser, identityNo)) {
                sendErrorResp(sender, message, identityNo, upMsgWrap, DownMsgErrorCode.TO_USER_BAN, packetFactory);
                return;
            } // end permission if

            if (!senderUser.getAppKey().equals(recvUser.getAppKey())) {
                sendErrorResp(sender, message, identityNo, upMsgWrap, DownMsgErrorCode.USER_APP_NOT_MATCH, packetFactory);
                return;
//                String wrap = DownWrapperUtil.newC2CWrap(packetFactory.newPacket(), identityNo, DownMsgErrorCode.USER_NOT_EQUAL_APP);
//                mqSender.syncSend(DownTopicConst.COMMON_TOPIC, c2CMsg.getMqTags(), wrap);
            } // end if appKey equals

            log.info("[c2c] 已加载相关信息到消息, 【{}】向用户【{}】说：{}", senderUser.getNick(), recvUser.getNick(), message.getPayload());
            fillFromUserNicknameAndAvatarToThisMsg(message, senderUser);

            if (StringUtils.isBlank(message.getCid())) {
                message.setCid(message.getReceiverId());
                message.setConversationType(ConverTypeEnum.C2C.getCode());
            } // end if msg.cid isBlank

            if (QoSReceiveDaemonC2S.getInstance()
                .hasReceived(genUserMsgFingerChar(IdentityUtil.buildIdentityNo(senderUser.getUserId() + "",
                    upMsgWrap.getDeviceType()), message.getClientMsgId()))) {
                log.warn("[c2c] 已丢弃客户端【{}】id={}的重复消息：{}", senderUser.getNick(), message.getClientMsgId(), message.getPayload());
                return;
            }
//        if (cacheHelper.isRepeatMessage(fromUser.getAppNo() + COLON + fromUser.getUserNo(), msg.getClientMsgId())) {
//            return;
//        }
            //
            // seq
            Long maxSeq = cacheService.incrSeq(buildSequenceDTO(senderUser.getAppKey(),
                ConverUtils.generateCid(message.getSenderId(), message.getReceiverId(), false)));
            message.setSeq(maxSeq.intValue());

            // qos
//            qosPoolAdd(upMsgWrap, message, senderUser);
            // kafka生产消息，第1条需要1.3s  :因为client 发送时候会进行聚合，数据量很大就会很快。不然会等500ms. 2024-12-09
            // store msg
            storeMsg(message, senderUser);

            updConvers(senderUser.getAppKey(), message);
            // 转发消息
            PDISPATCHERS.execute(() -> {
                try {
                    transDownMsgHandler.dispatchMsg(upMsgWrap, null, senderUser, recvUser, message);
                } catch (Exception var2) {
                    var2.printStackTrace();
                    log.error("[DISPATCHER-P] ======单聊转发器出现问题了。。。==>{}", var2.getMessage());
                }
            });

            log.info("[c2c] 【{}】向用户【{}】发送的单聊消息.处理完毕. use:{}ms", senderUser.getNick(), recvUser.getNick(), timer.interval());
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    private void updConvers(String appKey, Message message) {
        updConver(appKey, message, false);
        updConver(appKey, message, true);
    }

    private void updConver(String appKey, final Message message, boolean isRevert) {

        String senderId = message.getSenderId();
        String receiverId = message.getReceiverId();
        if (isRevert) {
            senderId = message.getReceiverId();
            receiverId = message.getSenderId();
        }

        final ConverItem converMe = ConverCache.get(ConverUtil.genPrivateConverKey(appKey, senderId, receiverId));
        if (null == converMe) {
            ConversationDTO conversation = converService.getConverFromCacheOrDB(appKey, senderId, receiverId);
            if (null == conversation) {
                // Todo
                // 创建会话
                // 保存会话
                conversation = new ConversationDTO();
                conversation.setIfMute(false);
                conversation.setIfTop(false);
            }
            ConverItem needAdd = new ConverItem();

            needAdd.setAppKey(appKey);

            needAdd.setConverType(1);
            needAdd.setUserId(senderId);
            needAdd.setTargetId(receiverId);

            needAdd.setIfMute(conversation.getIfMute());
            needAdd.setIfTop(conversation.getIfTop());

            needAdd.setTopUpdatedTime(conversation.getTopUpdatedTime());

            needAdd.setLatestMsgId(message.getId());
            needAdd.setLatestUnreadMsgIndex(message.getSeq());
            needAdd.setSyncTime(message.getTime());
            needAdd.setSortTime(message.getTime());
            if (conversation.getLatestUnreadMsgIndex() != null && conversation.getLatestUnreadMsgIndex() > message.getSeq()) {
                needAdd.setLatestMsgId(conversation.getLatestMsgId());
                needAdd.setLatestUnreadMsgIndex(conversation.getLatestUnreadMsgIndex());
                needAdd.setSyncTime(conversation.getSyncTime());
                needAdd.setSortTime(conversation.getSortTime());
                log.warn("[c2c] 更新会话出现问题，数据库数据为最新数据。");
            }

            needAdd.setLatestReadMsgId(conversation.getLatestReadMsgId());
            needAdd.setLatestReadMsgIndex(conversation.getLatestReadMsgIndex());
            ConverCache.insertOrUpd(ConverUtil.genPrivateConverKey(appKey, senderId, receiverId), needAdd);
//            syncHandler.handler(ConverUtil.genPrivateConverKey(appKey, senderId, receiverId));
            syncHandler.writeNow(ConverUtil.genPrivateConverKey(appKey, senderId, receiverId));
            return;
        }
        ConverItem updConver = converMe.clone();
        updConver.setLatestMsgId(message.getId());
        updConver.setLatestUnreadMsgIndex(message.getSeq());
        updConver.setSyncTime(message.getTime());
        updConver.setSortTime(message.getTime());
        ConverCache.insertOrUpd(ConverUtil.genPrivateConverKey(appKey, senderId, receiverId), updConver);
        syncHandler.writeNow(ConverUtil.genPrivateConverKey(appKey, senderId, receiverId));
//        syncHandler.handler(ConverUtil.genPrivateConverKey(appKey, senderId, receiverId));

//        ConverC2CUpdRequest request = new ConverC2CUpdRequest();
//        request.setAppKey(appKey);
//        request.setUserId(message.getSenderId());
//        request.setTargetId(message.getReceiverId());
//        request.setLatestMsg(message.getPayload());
//        request.setLatestMsgId(message.getId());
//        request.setLatestUnReadMsgIndex(message.getSeq());
//        request.setSortTime(message.getTime());
//        request.setSyncTime(message.getTime());
//        converService.updPrivateChatConvers(request);

    }

    protected boolean checkToUserPermission(UpMsgWrap c2CMsg, UserWrap fromUser, String identityNo) {
        if (UserStatusEnum.BAN.getCode() == fromUser.getStatus()) {
            return Boolean.TRUE;
        }
        return Boolean.FALSE;
    }

    /**
     * 将发送者昵称，头像，存入msg
     *
     * @param msg
     * @param sendUser
     */
    private void fillFromUserNicknameAndAvatarToThisMsg(Message msg, UserWrap sendUser) {
        // from用户头像，昵称，使用朋友表
        MsgUser user = new MsgUser();
        user.setNick(sendUser.getNick());
        user.setAvatar(sendUser.getAvatar());
        msg.setSenderInfo(user);

        FriendDTO friend = friendService.getFriendInfo(msg.getSenderId(), msg.getReceiverId());
        if (ObjectUtil.isNotNull(friend) && StringUtils.isNotBlank(friend.getNickname())) {
            user.setNick(friend.getNickname());
            msg.setSenderInfo(user);
        } // end if check null
    }

    private void qosPoolAdd(final UpMsgWrap wrapper, Message msg, UserWrap fromUser) {
        QoSReceiveDaemonC2S.getInstance()
            .addReceived(genUserMsgFingerChar(IdentityUtil.buildIdentityNo(fromUser.getUserId() + "",
                wrapper.getDeviceType()), msg.getClientMsgId()));

        // 这个是redis 做的去重。  防止客户端网络抖动，同时发送两次相同消息。   幂等
//        writeRedisCache(fromUser, msg.getClientMsgId());
    }

    private void storeMsg(final Message message, final UserWrap sender) {
        // mysql
        writeMessage(sender.getAppKey(), message);
    }

    protected void writeMessage(String appId, Message msg) {
        // save
        PrivateMsgReq privateMsg = new PrivateMsgReq();
        privateMsg.setAppKey(appId);
        privateMsg.setCid(ConverUtils.genFlexCid(msg.getSenderId(), msg.getReceiverId(), false));
        privateMsg.setSenderId(msg.getSenderId());
        privateMsg.setReceiverId(msg.getReceiverId());

        privateMsg.setSeq(msg.getSeq());

        privateMsg.setClientMsgId(msg.getClientMsgId());
        privateMsg.setMessageId(msg.getId());

        privateMsg.setPayload(msg.getPayload());
        privateMsg.setPayloadType(msg.getPayloadType());

        privateMsg.setType(msg.getType());

        privateMsg.setIfRecall(Boolean.FALSE);
        privateMsg.setBurnAfter(Boolean.FALSE);
        privateMsg.setIfRead(Boolean.FALSE);
        privateMsg.setSendTime(new Date(msg.getTime()));

        messageService.savePrivateMsg(privateMsg);
    }

    private void writeRedisCache(UserDTO fromUser, String clientMsgId) {
        cacheService.storeTempMsgId(fromUser.getAppKey() + COLON + fromUser.getUserId(), clientMsgId);
    }


    @NotNull
    private static List<String> genSearchUserDTO(Message msg) {
        return List.of(msg.getSenderId(), msg.getReceiverId());
    }


}
